package org.dbstar.scheduler.control;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

import org.dbstar.scheduler.beans.CmdJobConf;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.concurrent.MultiExecutorService;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.control.exception.SubmitException;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.job.IllegalFactoryException;
import org.dbstar.scheduler.job.JobRunner;
import org.dbstar.scheduler.job.JobStatus;
import org.dbstar.scheduler.job.SchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Job执行调度器
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-3-23
 */
public final class DefaultJobSchedulerExecutor extends Thread implements JobSchedulerExecutor {
	private static final Logger LOGGER = LoggerFactory.getLogger(DefaultJobSchedulerExecutor.class);

	private final MultiExecutorService executorService;
	private final JobController controller;
	private final SchedulerFactory schedulerFactory;
	private final long shutdownTimeout;

	private final Map<Future<Job>, JobRunner> executorMap;

	private volatile boolean shutdown = true;

	public DefaultJobSchedulerExecutor(MultiExecutorService executorService, JobController controller,
			SchedulerFactory schedulerFactory, long shutdownTimeout) {
		super(DefaultJobSchedulerExecutor.class.getName());
		this.executorService = executorService;
		this.controller = controller;
		this.schedulerFactory = schedulerFactory;
		this.shutdownTimeout = shutdownTimeout;
		this.executorMap = new ConcurrentHashMap<Future<Job>, JobRunner>();
		start();
	}

	@Override
	public synchronized Job submit(Identifier jobConfIdentifier, Map<String, String> parameters, String submitter)
			throws SubmitException, IllegalFactoryException {
		if (shutdown) {
			throw new SubmitException("JobSchedulerExecutor shutdown.");
		}
		try {
			return submit(schedulerFactory.createJob(jobConfIdentifier, parameters, submitter));
		} catch (DaoException e) {
			throw new SubmitException(e);
		}
	}

	private Job submit(Job job) throws DaoException {
		Identifier identifier = controller.addJob(job);
		LOGGER.info("Add job to controller succeed: {}", job);
		controller.changeJobStatus(identifier, JobStatus.submit);
		return controller.getJob(identifier);
	}

	@Override
	public synchronized boolean pause(Identifier jobIdentifier) throws DaoException {
		if (checkStatus(jobIdentifier, JobStatus.ready)) {
			controller.changeJobStatus(jobIdentifier, JobStatus.pause);
			return true;
		} else if (checkStatus(jobIdentifier, JobStatus.queue)) {
			if (cancelFuture(jobIdentifier, false)) {
				controller.changeJobStatus(jobIdentifier, JobStatus.pause);
				return true;
			} else {
				return false;
			}
		} else {
			return false;
		}
	}

	@Override
	public synchronized boolean resume(Identifier jobIdentifier) throws DaoException {
		if (checkStatus(jobIdentifier, JobStatus.pause)) {
			controller.changeJobStatus(jobIdentifier, JobStatus.ready);
			return true;
		} else {
			return false;
		}
	}

	@Override
	public synchronized boolean cancel(Identifier jobIdentifier) throws DaoException {
		if (checkStatus(jobIdentifier, JobStatus.ready, JobStatus.pause, JobStatus.init)) {
			controller.changeJobStatus(jobIdentifier, JobStatus.cancel);
			return true;
		} else if (checkStatus(jobIdentifier, JobStatus.queue)) {
			if (cancelFuture(jobIdentifier, false)) {
				controller.changeJobStatus(jobIdentifier, JobStatus.cancel);
				return true;
			} else {
				return false;
			}
		} else {
			return false;
		}
	}

	@Override
	public synchronized boolean kill(Identifier jobIdentifier) throws DaoException {
		if (checkStatus(jobIdentifier, JobStatus.run)) {
			return cancelFuture(jobIdentifier, true);
		} else {
			return false;
		}
	}

	private boolean checkStatus(Identifier jobIdentifier, JobStatus... jobStatuses) throws DaoException {
		Job job = controller.getJob(jobIdentifier);
		if (job != null) {
			JobStatus status = job.getStatus();
			for (JobStatus s : jobStatuses) {
				if (s == status) {
					return true;
				}
			}
		}
		return false;
	}

	private Future<Job> findFuture(Identifier jobIdentifier) {
		for (Entry<Future<Job>, JobRunner> entry : executorMap.entrySet()) {
			if (entry.getValue().getJob().getIdentifier().equals(jobIdentifier)) {
				return entry.getKey();
			}
		}
		return null;
	}

	private boolean cancelFuture(Identifier jobIdentifier, boolean mayInterruptIfRunning) {
		Future<Job> future = findFuture(jobIdentifier);
		if (future != null && !future.isDone() && !future.isCancelled()) {
			return future.cancel(mayInterruptIfRunning);
		}
		return false;
	}

	@Override
	public synchronized void start() {
		shutdown = false;
		// 将所有run和queue的job都改为ready，恢复异常状态
		try {
			for (Job job : controller.searchJobs(null, null, null, JobStatus.run, null, null, null, null, 0,
					Integer.MAX_VALUE)) {
				controller.changeJobStatus(job.getIdentifier(), JobStatus.ready);
			}
			for (Job job : controller.searchJobs(null, null, null, JobStatus.queue, null, null, null, null, 0,
					Integer.MAX_VALUE)) {
				controller.changeJobStatus(job.getIdentifier(), JobStatus.ready);
			}
		} catch (DaoException e) {
			e.printStackTrace();
		}
		super.start();
	}

	@Override
	public void run() {
		LOGGER.info("JobRunnerExecutor job distribution started.");

		// 循环直至executorService shutdown
		while (!executorService.isShutdown()) {
			// 循环直至任务队列中没有状态为ready的任务
			Job job;
			while (!executorService.isShutdown()) {
				try {
					job = controller.getFirstReadyJob(executorService.getFullResources());
				} catch (DaoException e) {
					job = null;
					LOGGER.error("未知访问异常", e);
				}
				if (job == null) {
					break;
				}
				JobRunner jobRunner;
				try {
					jobRunner = schedulerFactory.createJobRunner(job);
				} catch (IllegalFactoryException e) {
					job.getResult().setErrorException(e.getClass().getName() + ": " + e.getLocalizedMessage());
					job.setLaunchTime(System.currentTimeMillis());
					job.setFinishTime(System.currentTimeMillis());
					try {
						controller.updateJob(job);
						controller.changeJobStatus(job.getIdentifier(), JobStatus.error);
					} catch (DaoException e1) {
						LOGGER.error("未知访问异常", e1);
					}
					LOGGER.error("Create job runner fail: " + job, e);
					jobRunner = null;
				}
				if (jobRunner != null) {
					try {
						Future<Job> future = executorService.submit(jobRunner);
						try {
							controller.changeJobStatus(job.getIdentifier(), JobStatus.queue);
						} catch (DaoException e1) {
							LOGGER.error("未知访问异常", e1);
						}
						executorMap.put(future, jobRunner);
						LOGGER.debug("Submit job to executorService Succeed: {}", job);
					} catch (RejectedExecutionException e) {
						LOGGER.debug("Submit job to executorService Rejected: {}", job);
						checkFinishJob();
						if (!CmdJobConf.JOB_TYPE.equals(job.getType())) {
							try {
								checkReady();
							} catch (DaoException e2) {
								LOGGER.error("未知访问异常", e2);
							}

						}
						// 提交任务到执行队列时被拒绝，等待1秒重试。
						try {
							sleep(1000);
						} catch (InterruptedException e1) {
						}
					}
				}
			}
			checkFinishJob();
			if (!executorService.isShutdown()) {
				try {
					checkReady();
					if (controller.getFirstReadyJob() != null) {
						continue;
					}
				} catch (DaoException e) {
					LOGGER.error("未知访问异常", e);
				}
				synchronized (controller) {
					try {
						controller.wait(10000);
					} catch (InterruptedException e) {
					}
				}
			}
		}

		LOGGER.info("JobRunnerExecutor job distribution finished.");
	}

	private void checkReady() throws DaoException {
		for (Job job : controller.getJobs(JobStatus.submit)) {
			if (job.isReady()) {
				controller.changeJobStatus(job.getIdentifier(), JobStatus.ready);
			}
		}
	}

	@SuppressWarnings("unchecked")
	private void checkFinishJob() {
		for (Object future : executorMap.keySet().toArray()) {
			if (((Future<Job>) future).isDone()) {
				executorMap.remove(future);
			}
		}
	}

	@Override
	public synchronized void shutdown() throws InterruptedException, DaoException {
		if (!shutdown) {
			if (!executorService.isShutdown()) {
				// 先尝试关闭任务队列，这时不能再增加新任务到队列中，原先已经在队列中的任务仍旧可以继续执行。
				executorService.shutdown();
				this.interrupt();
			}
			// 在设置的时间内，等待队列中的任务执行完毕。
			if (!executorService.isTerminated() && !executorService.awaitTermination(shutdownTimeout, TimeUnit.SECONDS)) {
				// 如果任务不能在指定时间内执行完毕，那么强制终止。
				List<Runnable> list = executorService.shutdownNow();
				// 转储未执行的任务
				for (Runnable task : list) {
					JobRunner jobRunner = executorMap.remove(task);
					LOGGER.warn("Recover job from ExecutorService to JobController: {}", jobRunner.getJob());
					controller.changeJobStatus(jobRunner.getJob().getIdentifier(), JobStatus.ready);
				}
				// 再次等待所有任务终止。
				executorService.awaitTermination(shutdownTimeout, TimeUnit.SECONDS);
			}
			checkFinishJob();
			shutdown = true;
		}
	}
}
